package network;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ServerSocketChannel;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.protobuf.ByteString;

import command.ICallback;
import command.ICommand;
import command.ICommandFactory;
import command.IResponse;
import command.protobuf.MessageBufProtoBuf;
import command.protobuf.MessageBufProtoBuf.*;
import command.protobuf.MessageBufProtoBuf.RpcMessage.Builder;
import config.Configure;
import config.Config.ServiceMode;

import server.CommandFactories;
import server.CommandFactories.CmdType;

/**
 * In client mode, it only provides service to communicate with servers. In
 * server mode, besides the service of client mode, it also monitors the server
 * address and port to provide services to the client.
 * 
 * @author xiafan68
 * 
 */
public class MessageService {
	public static final Logger logger = LoggerFactory
			.getLogger(MessageService.class);
	
	public static RpcMessage NULL_RESPONSE_MESSAGE = makeEmptyMessage();
	
	public static int RESPONSE_FLAG = 0x1;
	public static int COMMAND_FLAG = 0x2;
	public static int CLIENT_SERVER_COMM = 0x80000000;
	public static int RPC_EXCEPTION = 0x4;

	
	private AtomicLong messageIDGen = new AtomicLong(0);
	private ConcurrentMap<Long, ICallback> messageCallback = new ConcurrentHashMap<Long, ICallback>();
	private OutputConnectionPool outputConnPool = new OutputConnectionPool();
	private SocketThread internalRpcThreadSocket = null;
	private ServerSocket internalRpcSocket = null;
	private ServiceMode mode = ServiceMode.Client;
	
	private Executor executor = Executors.newCachedThreadPool();
	
	private static MessageService instance = new MessageService();
	
	public static MessageService instance() {
		return instance;
	}

	private MessageService() {
		// TODO
		CommandFactories.init();
	}
	
	public void initServer() {
		try {
			mode = ServiceMode.Server;
			internalRpcSocket = new ServerSocket();

			ServerSocketChannel serverChannel = ServerSocketChannel.open();
			internalRpcSocket = serverChannel.socket();
			internalRpcSocket.setReuseAddress(true);
			InetAddress listenAddr = InetAddress.getByName(Configure
					.getListenAddress());
			InetSocketAddress address = new InetSocketAddress(listenAddr,
					Configure.getListenPort());

			internalRpcSocket.bind(address);
			internalRpcThreadSocket = new SocketThread(internalRpcSocket,
					"inernal rpc");
			internalRpcThreadSocket.start();
			logger.info("MessageService started on" + address.toString());
		} catch (IOException e1) {
			logger.error("start server fails");
		}
	}

	private void initClient(InetAddress target, short port) {

	}

	/**
	 * send to command to the target nodes
	 * 
	 * @param cmd
	 * @param addrs
	 * @throws IOException
	 */
	public ICallback sendCommand(CmdType cmdType, byte[] cmd,
			List<InetAddress> addrs, short port) throws IOException {
		RpcMessage message = makeMessage(cmdType, cmd);
		ICallback callback = CommandFactories.getCommand(cmdType)
				.createCallback();
		callback.init(addrs);
		registerCallback(message.getHeader().getID(), callback);

		for (InetAddress addr : addrs) {
			sendMessage(message, addr, port);
		}
		return callback;
	}

	private RpcMessage makeMessage(CmdType cmdType, byte[] cmd) throws IOException {
		RpcMessage.Builder builder = RpcMessage.newBuilder();
		ByteString byteString = ByteString.copyFrom(cmd);
		int flag = COMMAND_FLAG;
		if (mode == ServiceMode.Client)
			flag |= CLIENT_SERVER_COMM;
		
		RpcMessage.Header header = RpcMessage.Header.newBuilder()
				.setCmdType(cmdType.ordinal()).setFlag(flag)
				.setID(messageIDGen.incrementAndGet()).build();
		RpcMessage message = RpcMessage.newBuilder().setContent(byteString)
				.setHeader(header).build();

		return message;
	}

	public void sendMessage(RpcMessage message, InetAddress addr, int port) {
		OutputTcpConnection conn = outputConnPool.getOutputTcpConnection(addr,
				port);
		if (conn == null) {

		} else {
			conn.write(message);
		}
	}

	public void processMessage(RpcMessage message, Socket socket) {
		ICommandFactory factory = CommandFactories
				.getCommand(CommandFactories.cmdTypes[message.getHeader().getCmdType()]);
		if (factory == null) {
			// TODO this is an error, maybe should be recorded
			return;
		}
		
		ByteArrayInputStream input = new ByteArrayInputStream(message
				.getContent().toByteArray());
		try {
			// TODO process as an response
			if ((message.getHeader().getFlag() & RESPONSE_FLAG) == RESPONSE_FLAG) {
				ICallback callback = messageCallback.get(message.getHeader().getID());
				if (callback != null) {
					IResponse response = factory.createResponse();
					response.read(input);
					callback.preprocess(response);
				}
			} else {
				// TODO process as an command
				ICommand cmd = factory.createCommand();
				cmd.setOrigin(message, socket);
				//cmd.invoke();
				executor.execute(new CommandTask(cmd));
				// should we invoke here?
			}
		} catch (IOException e) {

		}

	}

	public void registerCallback(long id, ICallback callback) {
		messageCallback.putIfAbsent(id, callback);
	}

	public void removeCallback(long id) {
		messageCallback.remove(id);
	}

	private static class SocketThread extends Thread {
		private final ServerSocket server;

		SocketThread(ServerSocket server, String name) {
			super(name);
			this.server = server;
		}

		public void run() {
			while (true) {
				try {
					Socket socket = server.accept();
					new InComingTcpConnection(socket).start();
				} catch (AsynchronousCloseException e) {
					// this happens when another thread calls close().
					logger.info("MessageService shutting down server thread.");
					break;
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
			}
		}

		void close() throws IOException {
			server.close();
		}
	}
	
	private static class CommandTask implements Runnable {
		private ICommand cmd = null;
		public CommandTask(ICommand cmd) {
			this.cmd = cmd;
		}
		
		@Override
		public void run() {
			cmd.invoke();
		}	
	}
	
	
	private static RpcMessage makeEmptyMessage() {
		RpcMessage.Header header = RpcMessage.Header.newBuilder().setID(-2).setCmdType(CmdType.nullresponse.ordinal()).setFlag(RESPONSE_FLAG).build();
		return RpcMessage.newBuilder().setHeader(header).setContent(ByteString.EMPTY).build();
	}
}
